#include <sstream>
#include <algorithm>
#include <memory>
#include "config.hpp"
#include "mqtt.hpp"
#if ENABLE_MQTT == 1
#include "filter.hpp"
#include "filterio.h"
#include "misc.h"
#include "sys_var.hpp"

void MQTT::cbs::
connected (const std::string &cause )
{
	if( ptOwner ){
		if( cause.find( "onSuccess" ) != std::string::npos )
			ptOwner->OnConnected( MQTT::OK);
		else if( cause.find( "reconnect" ) != std::string::npos )
			ptOwner->OnConnected( MQTT::ERR_RECONNET );
		else{
			__show_line_1( "%s%s%s" , TRED , "unexpected connect cause." , TNORMAL );
		}
	}else{
		__show_line_1( "%s%s%s" , TRED , "Error of MQTT object.");
	}
}

void MQTT::cbs::
connection_lost (const std::string &cause)
{
	__show_line_1( "" , TRED , "connection lost, try to reconnect" , TNORMAL );
	if( ptOwner ){
		//ptOwner->open();
	}else{
		__show_line_1( "%s%s%s" , TRED , "Error of MQTT object." , TNORMAL );
	}
}

void MQTT::cbs::
message_arrived ( mqtt::const_message_ptr msg)
{
	mqtt::binary d = msg->get_payload();
	const char * data = d.data();
	size_t len = d.size();
	std::string t = msg->get_topic();
	if( data ){
		if( ptOwner )
			ptOwner->OnRecv( t , len , data , filterIO::OK );
		else{
			__show_line_1( "%s%s%s" , TRED , "Error of MQTT object." , TNORMAL );
		}
	}else{
		__show_line_1( "%s%s%s" , TRED , "Empty data." , TNORMAL );
	}
}

void MQTT::cbs::
delivery_complete (mqtt::delivery_token_ptr tok)
{
	if( ptOwner ){
		ptOwner->OnSend( MQTT::OK );
	}else{
		__show_line_1( "%s%s%s" , TRED , "Empty data." , TNORMAL );
	}
}

////////////////////////////////////////////////////////////////////////////////////////////////////
MQTT::MQTT( const std::string& url , int port , const std::string& uid , filterIO::io_type t , filterList * l )
:filterIO( t , l )
{
	m_url = url;
	m_port = port;

	std::stringstream  ss;
	ss << url << ":" << port;

	m_uri = ss.str();
}

MQTT::MQTT( const std::string& uri , const std::string& uid , filterIO::io_type t , filterList * l )
:filterIO( t , l )
{
	if( uri.find( ":" ) == std::string::npos ){
		m_port = -1;
		m_url = uri;
	}else{
		m_url = uri.substr( 0 , uri.find(":") );
		std::stringstream ss;
		ss << uri.substr( uri.find(":") + 1 , uri.length() );
		ss >> m_port;
	}

	m_uri = uri;
}

MQTT::~MQTT()
{
	close();
}

bool MQTT::on_forward( size_t len , const char * data , err_code &err , size_t& rlen , char ** rd )
{
	bool ret = false;
	if( data == nullptr ) return ret;
	if( len == 0 ) return ret;

	for( ArrayString::iterator it = m_topics.begin(); it != m_topics.end(); it ++ ){
		m_client->publish( *it , data , len , m_qos , false );
	}

	ret = true;
	rlen = len;
	*rd = ( char *)data;

	return ret;
}

bool MQTT::on_backward( size_t len , const char * data , err_code &err , size_t& rlen , char ** rd  )
{
	if( data == nullptr ) return false;
	if( len == 0 ) return false;
	rlen = len;
	*rd = ( char *)data;

	if( cb_recv ){

	}
	return true;
}

MQTT::err_code 
MQTT::open()
{
	err_code ret = OK;

	m_client = std::make_unique< mqtt::async_client >( m_uri , m_uid );

	if( m_client ){
		m_client->connect( m_opts );
		for( size_t i = 0; i < m_topics.size(); i ++ )
		m_client->subscribe( m_topics[ i ] , m_qos );
	}
	return ret;
}

MQTT::err_code 
MQTT::open( const std::string& url , int port )
{
	if( m_client ) return ERR_MQTT_ALREADY_CONNECTED;

	m_url = url;
	m_port = port;
	std::stringstream ss;
	ss << url << ":" << port;
	m_uri = ss.str();

	m_client = std::make_unique< mqtt::async_client >( m_uri , m_uid );
	if( m_client ){// 初始化完成，订阅预先设置的主题
		m_client->connect( m_opts );
		for( size_t i = 0; i < m_topics.size(); i ++ )
		m_client->subscribe( m_topics[ i ] , m_qos );
	}

	return OK;
}

MQTT::err_code 
MQTT::open( const std::string& uri )
{
	if( m_client ) return ERR_MQTT_ALREADY_CONNECTED;

	if( uri.find( ":" ) == std::string::npos ){
		m_port = -1;
		m_url = uri;
	}else{
		m_url = uri.substr( 0 , uri.find(":") );
		std::stringstream ss;
		ss << uri.substr( uri.find(":") + 1 , uri.length() );
		ss >> m_port;
	}

	m_uri = uri;
	m_client = std::make_unique< mqtt::async_client >( m_uri , m_uid );
	if( m_client ){// 初始化完成，订阅预先设置的主题
		m_client->connect( m_opts );
	}

	return OK;
}

void MQTT::close()
{
	if( m_client )
		m_client.reset();
}

MQTT& MQTT::login( const std::string& user , const std::string& pswd , bool en )
{
	m_user = user;
	m_pswd = pswd;

	m_en_log = en;

	m_opts.set_user_name( user );
	m_opts.set_password( pswd.c_str() );
	return *this;
}

MQTT& MQTT::uid( const std::string& uid )
{
	m_uid = uid;
	return *this;
}

MQTT& MQTT::dir( const std::string& dir )
{
	m_dir = dir;
	return *this;
}

MQTT& MQTT::sub( const std::string& topic )
{
	m_topics.push_back( topic );

	return *this;
}

MQTT& MQTT::dissub( const std::string& topic )
{
	ArrayString::iterator it = std::find( m_topics.begin() , m_topics.end() , topic );
	if( it != m_topics.end() ){
		m_topics.erase( it );

		if( m_client )
		m_client->unsubscribe( topic );
	}

	return *this;
}

MQTT& MQTT::Qos( int qos )
{
    	m_qos = qos;
    	return *this;
}

void MQTT :: init_sub()
{
	std::shared_ptr< CSysVar > ptvar = GetOrCreateSysVar();
	if( ptvar ){
		m_pub_topics.erase( m_pub_topics.begin() , m_pub_topics.end() );
		for( size_t i = 0; i < 10; i ++ ){
			std::stringstream	ss;
			ss << "/mqtt/sub" << i;
			std::string 		str;
			ptvar->GetValue( ss.str() , str );

			if( str.empty() == false ){
				m_sub_topics.push_back( str );
			}
		}
	}else{
		__show_line( "%s%s%s" , TRED , "Can not load system variable module" , TNORMAL );
	}
}

void MQTT :: init_pub()
{
	std::shared_ptr< CSysVar > ptvar = GetOrCreateSysVar();
	if( ptvar ){
		m_pub_topics.erase( m_pub_topics.begin() , m_pub_topics.end() );
		for( size_t i = 0; i < 10; i ++ ){
			std::stringstream	ss;
			ss << "/mqtt/pub" << i;
			std::string 		str;
			ptvar->GetValue( ss.str() , str );

			if( str.empty() == false ){
				m_pub_topics.push_back( str );
			}
		}
	}else{
		__show_line( "%s%s%s" , TRED , "Can not load system variable module" , TNORMAL );
	}
}

void MQTT :: init_params()
{
	std::shared_ptr< CSysVar > ptvar = GetOrCreateSysVar();
	try{
		if( ptvar ){
			ptvar->GetValue( "/mqtt/user" , m_user );
			ptvar->GetValue( "/mqtt/password" , m_pswd );
			ptvar->GetValue( "/mqtt/cid" , m_cid );
			ptvar->GetValue( "/mqtt/user" , m_user );
			bool data;
			ptvar->GetValue( "/mqtt/anony" , data );
			m_anony = data;
			ptvar->GetValue( "/mqtt/enable" , data );
			m_enable = data;
			ptvar->GetValue( "/mqtt/useCid" , data );
			m_useCid = data;
			ptvar->GetValue( "/mqtt/useAccount" , data );
			m_useAccount = data;
		}else{
			__show_line( "%s%s%s" , TRED , "Can not load system variable module" , TNORMAL );
		}
	}catch( ... ){
		__show_line( "%s%s%s" , TRED , "Pls check if missing system variable." , TNORMAL );
	}
}

void MQTT :: OnRecv( const std::string& topic , size_t len , const char * data , err_code e )
{
	if( data != nullptr && len != 0 ){
		filterIO::OnRecv( len , data , e );
		
		if( cb_recv ){
			cb_recv( topic , data , len );
		}
	}
}

MQTT& MQTT::OnRecv( std::function< > fun )
{
	cb_recv = fun;
}

void MQTT:: do_sub()
{
	if( !m_client ){
		return;
	}
	for( auto a : m_sub_topics ){
		m_client->subscribe( a , m_qos )
	}
}

void MQTT :: OnConnected( err_code e )
{
	if( e == OK ){
		do_sub();
	}

	if( cb_conn )
		cb_conn( e );
}
#endif